package me.shiqiang.im.consumer;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.fasterxml.jackson.core.JsonProcessingException;
import me.shiqiang.im.enums.MessageCType;
import me.shiqiang.im.enums.MessageCmd;
import me.shiqiang.im.msg.MsgBody;
import me.shiqiang.im.msg.NewMsg;
import me.shiqiang.im.redis.RedisKey;
import me.shiqiang.im.service.ChatSwitchService;
import me.shiqiang.im.service.MsgService;
import me.shiqiang.im.service.ServerNodeClient;
import me.shiqiang.im.utils.JsonUtils;
import me.shiqiang.im.util.RedisPipelineProcessor;
import io.socket.client.Socket;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @author cuishiqiang
 * @date 2017/12/16 下午3:56
 */
@Component(MessageCType.GROUP)
public class GroupMsgListener implements MsgListener {
    private Logger logger = LoggerFactory.getLogger(this.getClass());


    @Value("${send.topic}")
    private String topic;

    @Autowired
    MsgService service;

    @Value("${server.id}")
    String serverId;

    @Autowired
    ServerNodeClient serverNodeClient;

    @Autowired
    ChatSwitchService chatSwitchService;

    @Autowired
    Socket socket;

    @Autowired
    ConcurrentLinkedQueue<RedisPipelineProcessor> queue;

    @Override
    public Action consume(Message message, ConsumeContext context) {
        String msgBodyStr = new String(message.getBody());
        try {
            logger.info("群收到消息: {}", msgBodyStr);
            MsgBody msgBody = JsonUtils.getInstance().fromJson(msgBodyStr, MsgBody.class);
            Long groupId = Long.parseLong(message.getUserProperties("to"));
            Long syncKey = Long.parseLong(message.getUserProperties("syncKey"));
            queue.add(connection -> {
                try {
                    Set<byte[]> userIdBytes = connection.sMembers(RedisKey.groupUserSet(groupId).getBytes());
                    for (byte[] userIdByte : userIdBytes) {
                        String userId = new String(userIdByte);
                        if (!chatSwitchService.chatSwitch(Long.valueOf(userId))) {
                            return;
                        }
                        NewMsg fromNewMsg = new NewMsg(Long.parseLong(userId), syncKey, msgBody);
                        connection.zAdd(RedisKey.clientMsgQueue(userId).getBytes(), syncKey, msgBodyStr.getBytes());
                        socket.emit(MessageCmd.NEW_MSG, new JSONObject(JsonUtils.getInstance().toJson(fromNewMsg)));
                    }
                } catch (JsonProcessingException e) {
                    logger.error("转换异常", e);
                }
            });
            return Action.CommitMessage;
        } catch (Exception e) {
            logger.error(String.format("消息处理失败：%s", msgBodyStr), e);
            return Action.ReconsumeLater;
        }
    }
}
